package com.zm.xiaohashu.count.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONUtil;
import com.github.phantomthief.collection.BufferTrigger;
import com.zm.xiaohashu.count.biz.constant.MQConstants;
import com.zm.xiaohashu.count.biz.constant.RedisKeyConstants;
import com.zm.xiaohashu.count.biz.enums.LikeUnlikeNoteTypeEnum;
import com.zm.xiaohashu.count.biz.model.dto.LikeUnlikeNoteMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.security.PublicKey;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_NOTE_LIKE_OR_UNLIKE_COUNT,
        topic = MQConstants.TOPIC_NOTE_LIKE_OR_UNLIKE_COUNT)
public class CountLikeUnLikeConsumer implements RocketMQListener<String> {
    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000) // 缓存队列的最大容量
            .batchSize(1000)   // 一批次最多聚合 1000 条
            .linger(Duration.ofSeconds(1)) // 多久聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;


    @Override
    public void onMessage(String body) {
        log.info("## 消费到了 MQ 【计数: 点赞数/取消点赞数】, {}...", body);
        bufferTrigger.enqueue(body);
    }

    private void consumeMessage(List<String> strings) {
        if (CollUtil.isEmpty(strings)) return;
        List<LikeUnlikeNoteMqDTO> countLikeUnlikeNoteMqDTOS = strings.stream().map(body -> JSONUtil.toBean(body, LikeUnlikeNoteMqDTO.class)).toList();
        Map<Long, List<LikeUnlikeNoteMqDTO>> noteMap = countLikeUnlikeNoteMqDTOS.stream().collect(Collectors.groupingBy(LikeUnlikeNoteMqDTO::getNoteId));
        Map<Long, List<LikeUnlikeNoteMqDTO>> userMap = countLikeUnlikeNoteMqDTOS.stream().collect(Collectors.groupingBy(LikeUnlikeNoteMqDTO::getCreatorId));

        Map<Long, Integer> countNoteMap = getCountMap(noteMap);
        Map<Long, Integer> countUserMap = getCountMap(userMap);
        //更新redis
        //笔记维度
        countNoteMap.forEach((noteId, totalCount) -> {
            String countNoteRedisKey = RedisKeyConstants.buildCountNoteKey(noteId);
            boolean isExisted = redisTemplate.hasKey(countNoteRedisKey);
            if (isExisted) {
                redisTemplate.opsForHash().increment(countNoteRedisKey, RedisKeyConstants.FIELD_NOTE_LIKE_TOTAL, totalCount);
            }
        });

        //用户维度
        countUserMap.forEach((userId, totalCount) -> {
            String userCountKey = RedisKeyConstants.buildCountUserKey(userId);
            boolean isExisted = redisTemplate.hasKey(userCountKey);
            if (isExisted) {
                redisTemplate.opsForHash().increment(userCountKey, RedisKeyConstants.FIELD_NOTE_LIKE_TOTAL, totalCount);
            }
        });
        //mq落库
        sendMessage(countNoteMap, countUserMap);

    }

    private void sendMessage(Map<Long, Integer> countNoteMap, Map<Long, Integer> countUserMap) {
        Message<Map<Long, Integer>> noteMessage = MessageBuilder.withPayload(countNoteMap).build();
        String noteDestination = MQConstants.TOPIC_NOTE_LIKE_OR_UNLIKE_2_DB;
        rocketMQTemplate.asyncSend(noteDestination, noteMessage, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：点赞数入库】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：点赞数入库】MQ 发送异常: ", throwable);
            }
        });
        Message<Map<Long, Integer>> userMessage = MessageBuilder.withPayload(countUserMap).build();
        String userDestination = MQConstants.TOPIC_USER_LIKE_OR_UNLIKE_2_DB;
        rocketMQTemplate.asyncSend(userDestination, userMessage, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：用户点赞数入库】MQ 推送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：用户点赞数入库】MQ 推送异常: ", throwable);
            }
        });
    }

    public Map<Long, Integer> getCountMap(Map<Long, List<LikeUnlikeNoteMqDTO>> map) {
        Map<Long, Integer> noteCountMap = new HashMap<>();
        for (Map.Entry<Long, List<LikeUnlikeNoteMqDTO>> entry : map.entrySet()) {
            Long targetId = entry.getKey();
            List<LikeUnlikeNoteMqDTO> list = entry.getValue();
            int totalCount = 0;
            for (LikeUnlikeNoteMqDTO likeUnlikeNoteMqDTO : list) {
                Integer type = likeUnlikeNoteMqDTO.getType();
                totalCount += Objects.equals(type, LikeUnlikeNoteTypeEnum.LIKE.getCode()) ? 1 : -1;
            }
            noteCountMap.put(targetId, totalCount);
        }
        return noteCountMap;
    }
}
